//Job 的 DTO
package com.mr.job;

import java.io.File;

import com.mr.core.Combiner;
import com.mr.core.HashPartitioner;
import com.mr.core.Mapper;
import com.mr.core.Partitioner;
import com.mr.core.Reducer;

public class Job<K1, V1, K2, V2, K3, V3> {
    private String inputPath;
    private Class<? extends Mapper<K1, V1, K2, V2>> mapperClass;
    private Class<? extends Reducer<K2, V2, K3, V3>> reducerClass;
    private int numReduceTasks = 1;
    
    private Class<? extends Partitioner> partitionerClass = HashPartitioner.class;
    private Class<? extends Combiner<K2, V2>> combinerClass;
    
    private String intermediateDir = "tmp/";
    private String outputDir = "output/";

    public String getInputPath() { return inputPath; }
    public void setInputPath(String inputPath) { this.inputPath = inputPath; }
    
    public int getNumReduceTasks() { return numReduceTasks; }
    public void setNumReduceTasks(int numReduceTasks) { this.numReduceTasks = numReduceTasks; }
    public Class<? extends Partitioner> getPartitionerClass() {
        return partitionerClass;
    }
    
    public void setPartitionerClass(Class<? extends Partitioner> partitionerClass) {
        this.partitionerClass = partitionerClass;
    }
    
    public Class<? extends Mapper<K1, V1, K2, V2>> getMapperClass() { return mapperClass; }
    public void setMapperClass(Class<? extends Mapper<K1, V1, K2, V2>> mapperClass) { this.mapperClass = mapperClass; }
    
    public Class<? extends Reducer<K2, V2, K3, V3>> getReducerClass() { return reducerClass; }
    public void setReducerClass(Class<? extends Reducer<K2, V2, K3, V3>> reducerClass) { 
        this.reducerClass = reducerClass;
    }
    
    public Class<? extends Combiner<K2, V2>> getCombinerClass() {
        return combinerClass;
    }
    
    public void setCombinerClass(Class<? extends Combiner<K2, V2>> combinerClass) {
        this.combinerClass = combinerClass;
    }
    
    // 重试次数配置，对并发死锁进行处理
    private int maxRetry = 3;
    
    // 配置方法
    public int getMaxRetry() {
        return maxRetry;
    }
    
    public void setMaxRetry(int maxRetry) {
        this.maxRetry = maxRetry;
    }

    // 错误处理
    public String getIntermediateDir() { 
        File dir = new File(intermediateDir);
        if (!dir.exists() && !dir.mkdirs()) {
            throw new RuntimeException("Failed to create intermediate directory: " + intermediateDir);
        }
        return intermediateDir;
    }
    
    // 输出目录设置方法
    public void setOutputDir(String outputDir) {
        this.outputDir = outputDir;
    }

    public String getOutputDir() {
        // 目录自动创建
        new java.io.File(outputDir).mkdirs();
        return outputDir; 
    }
    
    // 线程数配置字段
    private int mapThreads = 1;
    private int reduceThreads = 1;

    public int getMapThreads() {
        return mapThreads;
    }

    public void setMapThreads(int mapThreads) {
        this.mapThreads = mapThreads;
    }

    public int getReduceThreads() {
        return reduceThreads;
    }

    public void setReduceThreads(int reduceThreads) {
        this.reduceThreads = reduceThreads;
    }
}